package fox.framework.rpc.endpoint.exchange;


import fox.framework.rpc.client.message.request.RpcClientRequest;
import fox.framework.rpc.client.message.response.RpcClientResponse;
import fox.framework.rpc.common.ExceptionUtils;
import fox.framework.rpc.common.StringUtils;
import fox.framework.rpc.exception.RpcRequestTimeOutException;
import fox.framework.rpc.exception.RpcRequestUndoException;
import fox.framework.rpc.message.RpcHeader;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author cuichao
 * @Description: BlockingQueueExchange
 * @Date: create in 2020/12/17 13:19
 */
public class BlockingQueueExchange implements RpcExchange {

    private final ConcurrentMap<String, Sync> selector;

    public BlockingQueueExchange() {
        selector = new ConcurrentHashMap<>();
    }

    @Override
    public Future register(RpcClientRequest request) {
        String requestId = request.getHeader(RpcHeader.REQUEST_ID);
        Sync sync = new Sync(requestId);
        selector.put(requestId, sync);
        return sync;
    }


    @Override
    public void exchange(RpcClientResponse response) {
        String requestId = response.getHeader(RpcHeader.REQUEST_ID);
        if (StringUtils.isEmpty(requestId)) {
            return;
        }
        Sync sync = selector.getOrDefault(requestId, null);
        if (sync != null) {
            sync.offer(response);
        }
    }


    private class Sync implements Future {
        //请求Id
        private String requestId;
        //同步列队
        private SynchronousQueue<RpcClientResponse> sync;
        /**
         * 同步器状态
         * 0 初始化
         * 1 完成响应
         * -1 取消请求
         */
        private AtomicInteger status = new AtomicInteger(0);
        //请求线程
        private volatile Thread requestThread;
        //取消线程异原因
        private volatile RpcRequestUndoException undoCause;


        public Sync(String requestId) {
            sync = new SynchronousQueue<>();
            this.requestId = requestId;
        }

        @Override
        public void undo(Throwable throwable){

            if(requestThread != null && status.compareAndSet(0,-1)){
                undoCause = new RpcRequestUndoException(throwable);
                selector.remove(this.requestId);
                requestThread.interrupt();
            }
        }

        private void offer(RpcClientResponse response){
           if(status.compareAndSet(0,1)){
               sync.offer(response);
           }
        }
        @Override
        public RpcClientResponse get() throws RpcRequestUndoException{
            requestThread = Thread.currentThread();
            RpcClientResponse response = null;
            try {
                response = sync.take();
            } catch (InterruptedException e) {
                if(isUndoStatus()){
                    throw undoCause;
                }
                throw ExceptionUtils.RpcRuntimeExceptionWrap(e);
            }finally {
                selector.remove(this.requestId);
            }
            selector.remove(this.requestId);
            return response;
        }

        @Override
        public RpcClientResponse get(long time, TimeUnit unit) throws InterruptedException {
            requestThread = Thread.currentThread();
            RpcClientResponse response = null;
            try {
                 response = sync.poll(time, unit);
            }catch (InterruptedException e){
                if(isUndoStatus()){
                    throw undoCause;
                }
                throw e;
            }finally {
                selector.remove(requestId);
            }
            selector.remove(requestId);
            if(response == null){
                throw new RpcRequestTimeOutException("time out");
            }
            return response;
        }

        private boolean isUndoStatus(){
            return status.get() == -1;
        }


    }




}
